
#include <stdio.h>
#include <errno.h>
#include <stdlib.h>
#include <errno.h>
#include <malloc.h>
#include <string.h>
#include <pthread.h>
#include <assert.h>
#include "Defines.h"
#include "Facade.h"
#include "panda.h"
#include "Handler.h"
#include "Debug.h"
#include "aclHandler.h"
/*
 * Worker thread: main event loop
 */
void * Thread_Worker_Start(void *arg) 
{
    LIBEVENT_THREAD *me = (LIBEVENT_THREAD *) arg;
    Facade *pR = Facade::Instance();
    /* Any per-thread setup can happen here; thread_init() will block until
     * all threads have finished initializing.
     */
    pR->GetThreads()->InitCountIncrease();
    return (void*) event_base_loop(me->base, 0);
}
/*
 * Processes an incoming "handle a new connection" item. This is called when
 * input arrives on the libevent wakeup pipe.
 */
void Thread_OnRead_Process(int fd, short which, void *arg) 
{
	LIBEVENT_THREAD *me = (LIBEVENT_THREAD *) arg;
	CQ_ITEM *item;
	char buf[1];
	Facade *pR = Facade::Instance();
	if (read(fd, buf, 1) != 1)
	{
		//  fprintf(stderr, "Can't read from libevent pipe\n");
	}
	item = pR->GetThreads()->cq_peek(&me->new_conn_queue);
	if (NULL != item) {
		conn *c = pR->GetNetWork()->conn_new(item->sfd, item->init_state, item->event_flags, item->read_buffer_size, item->is_udp, me->base);
		if (c == NULL) {
			if (item->is_udp) {
				fprintf(stderr, "Can't listen for events on UDP socket\n");
				exit(1);
			} else {
				if (pR->GetSettings()->verbose > 0) {
					fprintf(stderr, "Can't listen for events on fd %d\n",
							item->sfd);
				}
				close(item->sfd);
			}
		}
		pR->GetThreads()->cqi_free(item);
	}
}

/*
 * Sets a connection's current state in the state machine. Any special
 * processing that needs to happen on certain state transitions can
 * happen here.
 */
void Conn_Set_State(conn *c, int state) {
	DEBUG5(pthread_self()," set conn(", c->sfd,")'s state = ", state);
	assert(c != NULL);
	if (state != c->state) 
		c->state = state;
}

bool Update_Event(conn *c, const int new_flags) {
    DEBUG1(" ! ");
    assert(c != NULL);
    struct event_base *base = c->event.ev_base;
    if (c->ev_flags == new_flags)
        return true;
    if (event_del(&c->event) == -1) return false;
    event_set(&c->event, c->sfd, new_flags, Thread_Event_Handler, (void *)c);
    event_base_set(base, &c->event);
    c->ev_flags = new_flags;
    if (event_add(&c->event, 0) == -1) return false;
    return true;
}

/*
 * Sets whether we are listening for new connections or not.
 */
void Accept_New_Connection(const bool do_accept) {
	//conn *next;
	Facade *pR = Facade::Instance();
	if (! pR->GetThreads()->is_listen_thread())
		return;
       pR->GetThreads()->conn_listen_accept( do_accept );
}

void Handler_Dispatch(conn *c) {
    bool stop = false;
    int sfd, flags = 1;
   // struct sockaddr_storage addr;
    struct sockaddr_in addr;
    socklen_t addrlen = sizeof( addr );
   // int res;
    Facade *pR = Facade::Instance();
    aclHandler *pAcl = aclHandler::Instance();
    assert(c != NULL);
    while (!stop) {
	    switch(c->state) {
		    case conn_listening:
			    addrlen = sizeof(addr);
			    if ((sfd = accept(c->sfd, (struct sockaddr *)&addr, &addrlen)) == -1) {
				    if (errno == EAGAIN || errno == EWOULDBLOCK) {
					    /* these are transient, so don't log anything */
					    stop = true;
				    } else if (errno == EMFILE) {
                                            cerr<<"Too many connection ....."<< endl;
					    //Accept_New_Connection(false);
					    stop = true;
				    } else {
					    perror("accept()");
					    stop = true;
				    }
				    break;
			    }
			    DEBUG2("accept :", inet_ntoa(addr.sin_addr));
			    if(pAcl != NULL && !pAcl->AclCheck(inet_ntoa(addr.sin_addr)))
			    {
				    close(sfd);
				    ERRO<< "ip not in ACL list,can't accept it" <<endl;
                                    stop = true;
				    break;
			    }

			    if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 || fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) {
				    perror("setting O_NONBLOCK");
				    close(sfd);
				    break;
			    }
			    pR->GetThreads()->dispatch_conn_new(sfd, conn_read, EV_READ | EV_PERSIST, DATA_BUFFER_SIZE, false);
			    break;

        case conn_read:
	    if ( pR->GetNetWork()->try_read_network(c) > 0) 
            {
                    DEBUG2("read rbytes=", c->rbytes);
		    if(c->rbytes > 0)
		    { // printf("%s:\n%s\n", __func__, c->rcurr);
			    int rets = pR->GetHandler()->ReadProcess(c);
			    Conn_Set_State(c, rets);
			    if( rets == conn_closing)
				    break;
			    continue;
		    }
	    }
	    /* we have no command line and no data to read from network */
	    if (!Update_Event(c, EV_READ | EV_PERSIST)) {
                    DEBUG1("update event");
		    Conn_Set_State(c, conn_closing);
		    break;
	    }
	    stop = true;
            break;
	case conn_write:
	    if (c->wbytes > 0 ) 
	    {
		    if (pR->GetHandler()->WriteProcess(c) == conn_closing) 
		    {
			    Conn_Set_State(c, conn_closing);
			    break;
		    }
	    } 
	    /* fall through... */
	case conn_mwrite:
	    break;
        case conn_closing:
                pR->GetNetWork()->conn_close(c);
            stop = true;
            break;
        }
    }
  
    return;
}

void Thread_Event_Handler(const int fd, const short which, void *arg) {
    Facade *pR = Facade::Instance();
    conn *c;
    c = (conn *)arg;
    assert(c != NULL);
    c->which = which;
    /* sanity */
    if (fd != c->sfd) {
        pR->GetNetWork()->conn_close(c);
        return;
    }
     
    Handler_Dispatch(c);
    /* wait for next event */
    return;
}

